package com.stp.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListeners;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KafkaTestConsumer {
    /**
     * @SendTo
     * 在实际开发中，我们可能有这样的需求，应用A从TopicA获取到消息，经过处理后转发到TopicB，再由应用B监听处理消息，即一个应用处理完成后将该消息转发至其他应用，完成消息的转发
     * @param record
     */
    //@SendTo("sb_topic2")
    //@KafkaListener(topics = {"sb_topic"},errorHandler = "consumerAwareErrorHandler",containerFactory = "filterContainerFactory")
    @KafkaListener(topics = {"sb_topic"},errorHandler = "consumerAwareErrorHandler")
    public void onNormalMessage(ConsumerRecord<String, Object> record) {
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +
                record.value());
    }

    /**
     * 开启手动提交
     * @param record
     * @param ack
     */
    /*@KafkaListener(topics = {"sb_topic"},errorHandler = "consumerAwareErrorHandler")
    public void ackMessage(ConsumerRecord<String, Object> record, Acknowledgment ack) {
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" +
                record.value());
        //确认消息
        ack.acknowledge();
    }*/

    /**
     * 批量消费 yaml type: batch
     * 并且注意回调/监听器类型是否适配，不然报错
     */
    /*@KafkaListener(id = "consumer2", topics = {"sb_topic"}, groupId = "sb_group")
    public void onBatchMessage(List<ConsumerRecord<String, Object>> records) {
        System.out.println(">>> 批量消费一次，recoreds.size()=" + records.size());
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(record.value());
        }
    }*/


    /**
     * 转发收到的消息
     * @param record
     */
    @KafkaListener(topics = {"sb_topic2"})
    public void onMessage8(ConsumerRecord<?, ?> record) {
        System.out.println("收到sb_topic转发过来的消息：" + record.value());
    }


    /**
     * 指定多个消费者组。参数详解如上面的方法。
     *
     * @param record
     */
    /*@KafkaListeners({
            @KafkaListener(
                    groupId = APPLE_GROUP,
                    topicPartitions = {
                            @TopicPartition(topic = IPHONE_TOPIC, partitions = {"1", "2"}),
                            @TopicPartition(topic = IPAD_TOPIC, partitions = "1",
                                    partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "5"))
                    },
                    concurrency = "3"
            ),
            @KafkaListener(
                    groupId = XM_GROUP,
                    topicPartitions = {
                            @TopicPartition(topic = XMPHONE_TOPIC, partitions = {"1", "2"}),
                            @TopicPartition(topic = XMPAD_TOPIC, partitions = "1",
                                    partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "5"))
                    },
                    concurrency = "3"
            )
    }
    )
    public void groupIds(ConsumerRecord<String, String> record) {
        System.out.println("groupIds");
        System.out.println("内容：" + record.value());
        System.out.println("分区：" + record.partition());
        System.out.println("偏移量：" + record.offset());
        System.out.println("创建消息的时间戳：" + record.timestamp());
    }*/

}
